package com.hh.game.service.gateway.filter.statistics;

import io.netty.channel.*;

import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

@ChannelHandler.Sharable
public class StatisticsFilter extends ChannelInboundHandlerAdapter {
    private final ThreadGroup threadGroup = new ThreadGroup("statistics-work-thread-group");
    private AtomicInteger connections;
    private StatisticsManager statisticsManager;
    private Map<Long, TreeMap<Long, LongAdder>> qpsCache; //<threadID ,<second ,count>>
    private ScheduledExecutorService scheduled;

    public StatisticsFilter(int intervalTime ,StatisticsManager statisticsManager) {
        connections = new AtomicInteger(0);
        qpsCache = new ConcurrentHashMap<>();
        this.statisticsManager = statisticsManager;

        openStatisticsSaveJob(intervalTime);
    }
    private void openStatisticsSaveJob(int time){
        scheduled = new ScheduledThreadPoolExecutor(1,
                r -> new Thread(threadGroup ,r ,"statistics-work-thread"),
                (Runnable r, ThreadPoolExecutor executor) -> {
                });
        scheduled.scheduleAtFixedRate(() -> {
            StatisticsManager statistics = StatisticsFilter.this.statisticsManager;
            if(statistics != null) {
                StatisticsInfo info = new StatisticsInfo();
                info.setTime(System.currentTimeMillis());
                info.setConnctions(connections.get());
                info.setQps(countQps());
                statistics.save(info);

                qpsCache.clear();
            }
        } ,time ,time ,TimeUnit.SECONDS);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.connections.getAndIncrement();
        ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                StatisticsFilter.this.connections.getAndDecrement();
            }
        });
        super.channelActive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        long now = System.currentTimeMillis() / 1000;
        TreeMap<Long ,LongAdder> threadLongAdder = qpsCache.get(Thread.currentThread().getId());
        if(threadLongAdder == null){
            threadLongAdder = new TreeMap<>();
            qpsCache.put(Thread.currentThread().getId() ,threadLongAdder);
        }
        LongAdder longAdder = threadLongAdder.get(now);
        if(longAdder == null){
            longAdder = new LongAdder();
            threadLongAdder.put(now ,longAdder);
        }
        longAdder.increment();

        super.channelRead(ctx, msg);
    }

    private TreeMap<Long ,LongAdder> countQps(){
        LongAdder longAdder;
        TreeMap<Long ,LongAdder> count = new TreeMap<>();
        for(TreeMap<Long ,LongAdder> tmp : qpsCache.values()){
            for (Long time : tmp.keySet()){
                longAdder = count.get(time);
                if(longAdder == null){
                    longAdder = new LongAdder();
                    count.put(time ,longAdder);
                }
                longAdder.add(tmp.get(time).longValue());
            }
        }
        return count;
    }
}
